package kafkacli

import (
	"dc/qmodel"
	"encoding/json"
	"fmt"

	"github.com/Shopify/sarama"
	"github.com/gogf/gf/frame/g"
)

type KafkaProducer struct {
	conn sarama.SyncProducer
}

func NewKafkaProducer() (*KafkaProducer, error) {
	// to produce messages
	KafkaBrokers := g.Config().GetString("KafkaBrokers")
	fmt.Println("KafkaBrokers ", KafkaBrokers)

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForLocal        // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

	// 连接kafka
	client, err := sarama.NewSyncProducer(
		[]string{KafkaBrokers},
		config,
	)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return nil, err
	}

	return &KafkaProducer{conn: client}, nil
}

// CleanUp CleanUp
func (p *KafkaProducer) cleanUpProduce() {
	// if err := p.conn.Close(); err != nil {
	// 	log.Fatal("failed to close writer:", err)
	// }
}

// Produce Produce
func (p *KafkaProducer) produce(topic string, value string) error {

	// 构造一个消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic + getRandPrefix()
	msg.Value = sarama.StringEncoder(value)
	partition, offset, err := p.conn.SendMessage(msg)
	if err != nil {
		fmt.Println("send msg failed, err:", err)
		return nil

	}
	fmt.Printf("partition:%v offset:%v\n", partition, offset)

	return err
}

var (
	producer *KafkaProducer
)

// CleanUp CleanUp
func CleanUp() {
	producer.cleanUpProduce()
}

func init() {
	var err error
	producer, err = NewKafkaProducer()
	if err != nil {
		g.Log().Error("NewKafkaProducer err", err)
	}
}

func ProduceGetSubDomain(link *qmodel.Links) (err error) {

	res, _ := json.Marshal(link)
	err = producer.produce(TopicGetSubDomain, string(res))
	g.Log().Info("============>ProduceGetSubDomain link", g.Export(link.SubLink))
	return
}

func ProduceAllLink(link *qmodel.Links) (err error) {
	res, _ := json.Marshal(link)
	err = producer.produce(TopicAllLink, string(res))
	g.Log().Info("============>ProduceAllLink link", g.Export(link.SubLink))
	return

}

func ProduceDcgo(link *qmodel.Links) (err error) {
	res, _ := json.Marshal(link)
	err = producer.produce(TopicDcgo, string(res))
	g.Log().Info("============>ProduceDcgo link", g.Export(link.SubLink))
	return

}

func ProduceValidUrlCheck(link *qmodel.ValidUrlModel) (err error) {
	res, _ := json.Marshal(link)
	err = producer.produce(TopicValidUrlCheck, string(res))
	g.Log().Info("============>ProduceValidUrlCheck link", g.Export(link))
	return

}

func ProduceNewKeyword() (err error) {
	err = producer.produce(TopicNewKeyword, "")
	g.Log().Info("============>ProduceNewKeyword link")
	return

}
